package com.wsg.dep.core.event;


import com.wsg.dep.seq.FixedSequenceGroup;
import com.wsg.dep.seq.Sequence;
import com.wsg.dep.seq.Sequencer;

/**
 * 基于轮询的事件处理机制，用于从环形缓冲区（ring buffer）中拉取事件并处理
 */
public class EventPoller<T> {

    /**
     * 数据提供商
     */
    private final DataProvider<T> dataProvider;

    /**
     * 序列器，负责管理事件的发布和消费顺序。
     */
    private final Sequencer sequencer;

    /**
     * 当前 EventPoller 的消费序列，记录已经处理到的事件位置
     */
    private final Sequence sequence;

    /**
     * 门控序列，用于控制消费进度，确保不会消费到未发布的事件。
     */
    private final Sequence gatingSequence;


    public interface Handler<T> {
        /**
         * 事件处理
         * @param event 事件对象
         * @param sequence 事件的序列编号
         * @param endOfBatch 是否为批量事件的最后一个事件
         * @return 返回一个布尔值，表示是否继续处理下一个事件
         * @throws Exception
         */
        boolean onEvent(T event, long sequence, boolean endOfBatch) throws Exception;
    }

    /**
     * 轮询状态
     */
    public enum PollState {
        /**
         * 表示正在处理事件
         */
        PROCESSING,
        /**
         * 表示由于门控序列的限制，暂时无法处理更多事件
         */
        GATING,
        /**
         * 表示没有可用的事件可供处理
         */
        IDLE
    }

    /**
     * 构造函数
     * @param dataProvider 数据供应商
     * @param sequencer 序列器，负责管理事件的发布和消费顺序。
     * @param sequence 当前 EventPoller 的消费序列，记录已经处理到的事件位置
     * @param gatingSequence 门控序列，用于控制消费进度，确保不会消费到未发布的事件。
     */
    public EventPoller(
            final DataProvider<T> dataProvider,
            final Sequencer sequencer,
            final Sequence sequence,
            final Sequence gatingSequence) {
        this.dataProvider = dataProvider;
        this.sequencer = sequencer;
        this.sequence = sequence;
        this.gatingSequence = gatingSequence;
    }

    /**
     * 从指定的事件处理器中拉取数据
     * @param eventHandler
     * @return
     * @throws Exception
     */
    public PollState poll(final Handler<T> eventHandler) throws Exception {
        // 当前序列
        final long currentSequence = sequence.get();
        // 下一个序列
        long nextSequence = currentSequence + 1;
        // 获取可以安全地从环形缓冲区读取的最高序列编号。
        final long availableSequence = sequencer.getHighestPublishedSequence(nextSequence, gatingSequence.get());

        if (nextSequence <= availableSequence) {
            boolean processNextEvent;
            long processedSequence = currentSequence;

            try {
                do {
                    // 获取事件
                    final T event = dataProvider.get(nextSequence);
                    // 处理事件
                    processNextEvent = eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);
                    // 处理的序列
                    processedSequence = nextSequence;
                    // 下一个需要处理的序列
                    nextSequence++;

                }
                while (nextSequence <= availableSequence & processNextEvent);
            } finally {
                // 更新处理的序列位置
                sequence.set(processedSequence);
            }
            // 返回事件正在处理
            return PollState.PROCESSING;
        } else if (sequencer.getCursor() >= nextSequence) {
            return PollState.GATING;
        } else {
            return PollState.IDLE;
        }
    }

    /**
     * 轮询处理器的构造方法
     * @param dataProvider 数据提供商
     * @param sequencer 序列器，负责管理事件的发布和消费顺序。
     * @param sequence 当前 EventPoller 的消费序列，记录已经处理到的事件位置
     * @param cursorSequence 游标序列
     * @param gatingSequences 门控序列，用于控制消费进度，确保不会消费到未发布的事件。
     * @param <T>
     * @return
     */
    public static <T> EventPoller<T> newInstance(
            final DataProvider<T> dataProvider,
            final Sequencer sequencer,
            final Sequence sequence,
            final Sequence cursorSequence,
            final Sequence... gatingSequences) {
        Sequence gatingSequence;
        if (gatingSequences.length == 0) {
            gatingSequence = cursorSequence;
        } else if (gatingSequences.length == 1) {
            gatingSequence = gatingSequences[0];
        } else {
            gatingSequence = new FixedSequenceGroup(gatingSequences);
        }
        return new EventPoller<T>(dataProvider, sequencer, sequence, gatingSequence);
    }

    /**
     * 获取当前的消费序列
     * @return
     */
    public Sequence getSequence() {
        return sequence;
    }
}
